package net.bwie.jtp.dws.log.Function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.jtp.dws.log.bean.PageViewBean;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;


import java.util.HashMap;

public class PageViewBeanMapFunction extends RichMapFunction<String, PageViewBean> {

    //1-声明变量
    private transient ValueState<String> lastVisitDateState = null;
    //地区维度字典集合
    private HashMap<String, String> areaDic = null;



    //2-初始化
    @Override
    public void open(Configuration parameters) throws Exception {
//        /*
//       todo 每个mid使用状态保存最近一次访问日期，由于统计每天UV，所以状态生命周期为1天即可，
//        可以设置状态TTL为1day，节省内存使用
//     */
//        // 状态描述符
//        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>(
//                "lastVisitDate", String.class
//        );
//
//        // ttl配置
//        StateTtlConfig ttlConfig = StateTtlConfig
//                .newBuilder(Time.days(1))
//                .updateTtlOnCreateAndWrite()
//                .neverReturnExpired()
//                .build();
//        valueStateDescriptor.enableTimeToLive(ttlConfig);
//
//        // 2初始化
//        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);

        //1-初始化状态
        lastVisitDateState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("last-visit-date", String.class)
        );

        //初始化字典集合数据
        areaDic = getAreaDic();

    }

    private HashMap<String, String> getAreaDic() {
        // map集合
        HashMap<String, String> map = new HashMap<>();
        // 添加数据
        map.put("110000","北京");
        map.put("120000","天津");
        map.put("140000","山西");
        map.put("150000","内蒙古");
        map.put("130000","河北");
        map.put("310000","上海");
        map.put("320000","江苏");
        map.put("330000","浙江");
        map.put("340000","安徽");
        map.put("350000","福建");
        map.put("360000","江西");
        map.put("370000","山东");
        map.put("710000","台湾");
        map.put("230000","黑龙江");
        map.put("220000","吉林");
        map.put("210000","辽宁");
        map.put("610000","陕西");
        map.put("620000","甘肃");
        map.put("630000","青海");
        map.put("640000","宁夏");
        map.put("650000","新疆");
        map.put("410000","河南");
        map.put("420000","湖北");
        map.put("430000","湖南");
        map.put("440000","广东");
        map.put("450000","广西");
        map.put("460000","海南");
        map.put("810000","香港");
        map.put("820000","澳门");
        map.put("510000","四川");
        map.put("520000","贵州");
        map.put("530000","云南");
        map.put("500000","重庆");
        map.put("540000","西藏");
        // 返回集合
        return map ;
    }



    @Override
    public PageViewBean map(String value) throws Exception {

        //a-解析json数据
        JSONObject jsonObject = JSON.parseObject(value);
        JSONObject object = jsonObject.getJSONObject("common");
        JSONObject pageobject = jsonObject.getJSONObject("page");
        
//        通过地区编码获取地区名称
        String arValue = object.getString("ar");
        String province = areaDic.get(arValue);

        //uv访客数
        Long uvCount = 0L;
        //获取上一个访问日期
        String lastVisitDate = lastVisitDateState.value();
        //获取当前日期
        String visitDate = DateTimeUtil.convertLongToString(jsonObject.getLong("ts"), DateTimeUtil.DATE_FORMAT);

        //值为null时 表示为今日第一次访问uv=1
        //值不为null时 上一次访问日期与当前日期不同 也表示今日为第一次访问 uv=1
        if(lastVisitDate==null || !lastVisitDate.equals(visitDate)){
            uvCount = 1L;
            lastVisitDateState.update(visitDate);
        }
        return  new PageViewBean(
                null,
                null,
                object.getString("ba"),
                object.getString("ch"),
                province,
                object.getString("is_new"),
                pageobject.getString("last_page_id") == null ? 1L : 0L,
                pageobject.getLong("during_time"),
                1L,
                uvCount,
                jsonObject.getLong("ts")
        );
    }


    @Override
    public void close() throws Exception {
        //关闭资源
        super.close();
    }

}
